package cn.pm.datasource.sharding;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.IdUtil;
import cn.pm.datasource.sharding.bean.AutoCreateTablesCacheBean;
import cn.pm.datasource.sharding.bean.ShardingRedisMessage;
import cn.pm.datasource.sharding.consts.RedisListenerTopicConst;
import cn.pm.datasource.sharding.utils.LockUtil;
import com.alibaba.fastjson.JSONArray;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.shardingjdbc.jdbc.core.datasource.ShardingDataSource;
import org.junit.platform.commons.util.StringUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Create by spf on 2019/11/28
 */
@Slf4j
@Component
public class ShardingTableNameUtils {

    // 存储 逻辑表名--实际表名 的内存缓存集合
    private static final ConcurrentHashMap<String, AutoCreateTablesCacheBean> cache = new ConcurrentHashMap();

    // 数据源
    private static ShardingDataSource shardingDataSource;
    @Resource
    private void setShardingDataSource(ShardingDataSource shardingDataSource) {
        ShardingTableNameUtils.shardingDataSource = shardingDataSource;
    }
    // Redis连接  主要用于 在表更新时 发布广播消息
    private static RedisTemplate<Object, Object> redisTemplate;
    @Resource
    private void setRedisTemplate(RedisTemplate<Object, Object> redisTemplate) {
        ShardingTableNameUtils.redisTemplate = redisTemplate;
    }

    // 用以主动通知其他应用 数据库发生变更的url
    private static List<String> autoCreateTablesUpdateCacheUrl = new ArrayList<>();
    @Resource
    public void setAutoCreateTablesUpdateCacheUrl(ShardingConfig shardingConfig) {
        if (shardingConfig == null || shardingConfig.getAutoCreateTablesUpdateCacheUrl() == null || shardingConfig.getAutoCreateTablesUpdateCacheUrl().isEmpty())
            return;
        for (String url : shardingConfig.getAutoCreateTablesUpdateCacheUrl()) {
            if (StringUtils.isBlank(url))
                continue;
            url = url.trim();
            if (!url.startsWith("http"))
                continue;
            this.autoCreateTablesUpdateCacheUrl.add(url);
        }
        if (!autoCreateTablesUpdateCacheUrl.isEmpty())
            log.info("动态建表同步通知url：{}", autoCreateTablesUpdateCacheUrl);
    }

    /**
     * 向集合中添加表信息
     */
    public static void addTable(String logicTableName, AutoCreateTablesCacheBean bean) {
        if (logicTableName == null)
            return;
        cache.put(logicTableName, bean);
    }

    /**
     * 创建真实表
     */
    public static void createTable(String realTableName, AutoCreateTablesCacheBean bean) {
        int count = 1;
        String uuid = IdUtil.fastSimpleUUID();

        // 以分布式锁的形式保证建表有序执行，针对需要创建表的表名称加锁
        while (true) {
            if (LockUtil.lock(realTableName, uuid, 30)) {
                try {
                    // 查询真实表是否已存在
                    DataSource dataSource = shardingDataSource.getDataSourceMap().get(bean.getDataSourceName());
                    if (dataSource == null)
                        throw new Exception("数据源：" + bean.getDataSourceName() + " 不存在");
                    List<String> allTableNames = getAllTableNames(dataSource, realTableName, true);
                    if (allTableNames != null && !allTableNames.isEmpty())
                        return;

                    // 创建新的真实表
                    createTable(dataSource, bean.getLogicTableName(), realTableName);

                    // 更新此表的真实数据
                    List<String> newAllTableNames = getAllTableNames(dataSource, bean.getLogicTableName() + "_____%");
                    bean.setRealTableNameList(newAllTableNames);
                    cache.put(bean.getLogicTableName(), bean);

                    // 执行完毕 跳出循环
                    break;
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    LockUtil.unLock(realTableName, uuid);
                    //LockUtil.unlock(true);
                }
            }

            // 如果已经尝试加锁了10次  直接抛出异常即可
            if (count == 10) {
                log.error("创建真实表:" + realTableName + " 分布式锁尝试加锁20次仍不能成功，主动终止");
                break;
            }
            // 没到20次，休眠500毫秒后重试
            count++;
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 更新表缓存
     */
    public static void updateTable(String logicTableName) throws Exception {
        if (StringUtils.isBlank(logicTableName))
            return;
        AutoCreateTablesCacheBean table = cache.get(logicTableName);
        if (table == null)
            return;

        // 获取数据源
        DataSource dataSource = shardingDataSource.getDataSourceMap().get(table.getDataSourceName());
        if (dataSource == null)
            throw new Exception("数据源：" + table.getDataSourceName() + " 不存在");
        // 更新此表的真实数据
        List<String> newAllTableNames = getAllTableNames(dataSource, logicTableName + "_____%");
        table.setRealTableNameList(newAllTableNames);
        cache.put(logicTableName, table);
    }

    // 额外检查 指定的实际表是否已存在于缓存  如果已经存在 可以不用更新缓存
    public static void updateTable(String logicTableName, String realTableName, String dataSourceCatalog) throws Exception {
        if (StringUtils.isBlank(logicTableName))
            return;
        AutoCreateTablesCacheBean bean = cache.get(logicTableName);
        if (bean == null)
            return;

        // 校验本进程与广播的进程 是否用的同一个数据库
        if (StringUtils.isNotBlank(dataSourceCatalog)) {
            DataSource dataSource = shardingDataSource.getDataSourceMap().get(bean.getDataSourceName());
            if (dataSource == null)
                return;
            Connection connection = null;
            String catalog = null;
            try {
                connection = dataSource.getConnection();
                catalog = connection.getCatalog();
            } catch (Exception ignore) {
            }finally {
                closeResource(connection, null, null);
            }
            if (!dataSourceCatalog.equals(catalog))
                return;
        }

        // 校验目标表 是否已经存在了
        List<String> tableNameList = bean.getRealTableNameList();
        if (tableNameList != null && StringUtils.isNotBlank(realTableName) && tableNameList.contains(realTableName))
            return;
        updateTable(logicTableName);

        log.info("通过redis广播更新了实际表 {} -> {}",
                JSONArray.toJSONString(tableNameList),
                JSONArray.toJSONString(cache.get(logicTableName).getRealTableNameList()));
    }

    /**
     * 根据逻辑表名 获取实际表的相关信息
     */
    public static AutoCreateTablesCacheBean getTable(String logicTableName) {
        if (logicTableName == null)
            return null;
        return cache.get(logicTableName);
    }


    /**
     * 根据数据源，和表名规则 模糊查询所有表名
     */
    public static List<String> getAllTableNames(DataSource dataSource, String tableNameFormat) {
        return getAllTableNames(dataSource, tableNameFormat, false);
    }
    public static List<String> getAllTableNames(DataSource dataSource, String tableNameFormat, boolean equals) {
        if (dataSource == null)
            return null;
        List<String> result = new ArrayList<>();

        // 对于表名的判断符 使用=还是like
        String tableNameHyphen = equals ? "=" : "LIKE";
        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet resultSet = null;
        try {
            connection = dataSource.getConnection();
            ps = connection.prepareStatement("SELECT TABLE_NAME AS tableName FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_NAME " + tableNameHyphen + " ?");
            ps.setString(1, connection.getCatalog());
            ps.setString(2, tableNameFormat);
            resultSet = ps.executeQuery();
            if (resultSet != null) {
                while (resultSet.next()) {
                    String tableName = resultSet.getString("tableName");
                    result.add(tableName);
                }
            }
        } catch (Exception e) {
            log.error("查询所有表名失败 tableNameFormat:" + tableNameFormat, e);
        } finally {
            closeResource(connection, ps, resultSet);
        }
        return result;
    }


    /**
     * 创建真实表
     */
    private static void createTable(DataSource dataSource, String logicTableName, String realTableName) {
        if (dataSource == null)
            return;
        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet resultSet = null;
        String createTableSql = null;
        try {
            connection = dataSource.getConnection();
            ps = connection.prepareStatement("SHOW CREATE TABLE " + logicTableName);
            resultSet = ps.executeQuery();
            if (resultSet != null) {
                if (resultSet.next()) {
                    createTableSql = resultSet.getString(2);
                }
            }
        } catch (Exception e) {
            log.error("查询建表语句失败", e);
        } finally {
            closeResource(connection, ps, resultSet);
        }

        if (createTableSql == null) {
            log.warn("原始表 {} 建表语句查询为空 -> {}", logicTableName, realTableName);
            return;
        }
        // 转换创建表的语句 更换表名
        createTableSql = createTableSql.replaceAll("(?i)(CREATE TABLE `)[^`]*(`.*)", "$1" + realTableName + "$2");
        try {
            connection = dataSource.getConnection();
            ps = connection.prepareStatement(createTableSql);
            ps.execute();

            String dataSourceCatalog = connection.getCatalog();
            // 通知其他应用  真实表已经发生变更
            ThreadUtil.execute(() -> {
                // 1.通过redis广播的形式
                String token = IdUtil.simpleUUID();
                ShardingRedisMessage message = new ShardingRedisMessage();
                message.setToken(token);
                message.setDataSourceCatalog(dataSourceCatalog);
                message.setLogicTableName(logicTableName);
                message.setRealTableName(realTableName);
                ShardingRedisListener.flagToken.add(token);
                redisTemplate.convertAndSend(RedisListenerTopicConst.AUTO_CREATE_TABLE_UPDATE_TOPIC, message);

                // 2.通过url调用的形式
                if (!autoCreateTablesUpdateCacheUrl.isEmpty()) {
                    for (String url : autoCreateTablesUpdateCacheUrl) {
                        ThreadUtil.execute(() -> {
                            // 执行请求
                            Map params = new HashMap();
                            params.put("logicTableName", logicTableName);
                            //restTemplate.getForEntity(url + "?logicTableName={logicTableName}", Object.class, params);
                        });
                    }
                }
            });

        } catch (Exception e) {
            log.error("创建真实表失败", e);
        } finally {
            closeResource(connection, ps, resultSet);
        }
    }


    /**
     * JDBC释放资源
     */
    public static void closeResource(Connection connection, Statement statement, ResultSet resultSet) {
        if (resultSet != null) {
            try {
                resultSet.close();
            } catch (SQLException e) {
                log.error("JDBC resultSet 释放失败", e);
            }
        }
        if (statement != null) {
            try {
                statement.close();
            } catch (SQLException e) {
                log.error("JDBC statement 释放失败", e);
            }
        }

        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                log.error("JDBC connection 释放失败", e);
            }
        }
    }

}
